package manager

import (
	"fmt"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"github.com/cihub/seelog"

	"kyland.com/eagle/lib/cache"
	"kyland.com/eagle/lib/node"
)

func Schedule(job *AITask) {
	var mems []string
	Mutex.RLock()
	for mem := range node.NodesMap {
		mems = append(mems, mem)
	}
	Mutex.RUnlock()
	seelog.Infof("Begin to schedule, the healthy nodes: %v", mems)
	quantity := int64(len(mems))

	preselectedNodes := node.GetSufficientBandwidthNodes()
	// 如果只有一个节点或者节点带宽小于10M，则直接执行
	if quantity == 1 || len(preselectedNodes) == 0 {
		job.MaxSplit = 1
		job.Nodes = mems
		job.Node = node.Localhost
		job.Master = node.Localhost
		job.Src = job.Name
		job.Status = TaskCreated
		TaskSave(TaskPrefix, job)
		TaskSave(UnFinishedPrefix, job)
		ExecuteTask(job)
		job.Status = TaskFinished
		TaskSave(TaskPrefix, job)
		TaskDelete(UnFinishedPrefix, job)
	} else {
		seelog.Infof("Begin to schedule, the preselected node nums: %d", len(preselectedNodes))

		splitBegin := time.Now()
		job.SplitBegin = splitBegin.Format("2006-01-02 15:04:05")
		job.MaxSplit = 3
		if quantity <= job.MaxSplit {
			job.MaxSplit = quantity
		}

		subAiTask := []*AITask{}
		scheNodes := []string{}
		var part, calculate, split int64
		calculate, split = 0, job.Duration/job.MaxSplit
		scheNodes = append(scheNodes)

		paths, filename := filepath.Split(job.Name)
		src := fmt.Sprintf("%s%d-%s", paths, part, filename)
		destName := strings.Replace(src, "video", "result", 1)
		more := os.Getenv("VIDEO_MORE")
		moreInt64, err := strconv.ParseInt(more, 10, 64)
		if err != nil {
			fmt.Printf("Error: %s", err)
		}
		length := moreInt64
		sa := &AITask{
			Name:     job.Name,
			Master:   node.Localhost,
			Node:     node.Localhost,
			MaxSplit: job.MaxSplit,
			Status:   TaskCreated,
			Start:    0,
			End:      split + length,
			Src:      src,
			Dst:      destName,
		}

		calculate += split + length
		part += 1
		subAiTask = append(subAiTask, sa)
		scheNodes = append(scheNodes, node.Localhost)

		for _, mem := range preselectedNodes {
			if mem.Addr == node.Localhost {
				continue
			}

			scheNodes = append(scheNodes, mem.Addr)
			destName = strings.Replace(job.Name, "video", "result", 1)
			var end int64
			end = calculate + split
			if part == job.MaxSplit-1 {
				end = -1
			}

			src = fmt.Sprintf("%s%d-%s", paths, part, filename)
			destName = strings.Replace(src, "video", "result", 1)

			sa = &AITask{
				Name:     job.Name,
				Master:   node.Localhost,
				Node:     mem.Addr,
				MaxSplit: job.MaxSplit,
				Status:   TaskCreated,
				Start:    calculate,
				End:      end,
				Src:      src,
				Dst:      destName,
			}
			part += 1
			calculate += split + length
			subAiTask = append(subAiTask, sa)
			if int64(len(scheNodes)) == job.MaxSplit {
				break
			}
		}

		job.Nodes = scheNodes
		job.SubTask = subAiTask
		splitEnd := time.Now()
		job.SplitEnd = splitEnd.Format("2006-01-02 15:04:05")
		job.SplitCost = splitEnd.Unix() - splitBegin.Unix()
		_, finame := filepath.Split(job.Name)
		taskChange := fmt.Sprintf("%s-status", finame)
		cache.Cache.SetCache(taskChange, job)
		seelog.Infof("Begin to schedule %v", job)
		TaskSave(UnFinishedPrefix, job)
		TaskSave(TaskPrefix, job)

		for _, sub := range job.SubTask {
			tsk := sub
			go func() {
				seelog.Infof("begin to checkexecute%v", tsk)
				//tsk.Begin = time.Now().Format("2006-01-02 15:04:05")
				CheckExecute(tsk)
			}()
		}

		//cache.Cache.SetCache(taskChange, job)
		seelog.Infof("Begin to execute %v", job)
		ticker := time.NewTicker(10 * time.Second)
		done := make(chan struct{})
		finished := make(chan struct{})
		seelog.Infof("tick check begin %d", time.Now().Unix())
		go func() {
			finishedMap := make(map[string]int)
			for i := range job.SubTask {
				finishedMap[job.SubTask[i].Src] = TaskRunning
			}
			for {
				select {
				case <-done:
					return
				case t := <-ticker.C:
					seelog.Infof("Tick at: %d", time.Now().Unix())
					seelog.Infof("begin to check finishedMap:%v, %v", finishedMap, node.NodesMap)
					if checkTaskFinish(finishedMap, job) {
						finished <- struct{}{}
					}

					writeFlag := false
					for i, _ := range job.SubTask {
						seelog.Infof("Tick at checksub: %d", t.Second())
						status := TaskRunning
						//job.SubTask[i].Status = TaskRunning
						_, cname := filepath.Split(job.SubTask[i].Src)
						taskProcess := fmt.Sprintf("%s-process", cname)
						if job.SubTask[i].Node == job.SubTask[i].Master {
							proTime, exist := cache.Cache.GetCache(taskProcess)
							seelog.Infof("scheduleExecute1 taskPro:%s", taskProcess)
							if exist && job.SubTask[i].Finish != "" {
								tmp := proTime.(*AITask)
								if tmp.Begin != "" {
									job.SubTask[i].Begin = tmp.Begin
								}

								if tmp.Finish != "" {
									job.SubTask[i].Finish = tmp.Finish
								}

								if tmp.Cost != 0 {
									job.SubTask[i].Cost = tmp.Cost
								}

								cache.Cache.SetCache(taskChange, job)
								seelog.Infof("scheduleExecute1 taskPro:%s, %v, %v", taskProcess, tmp, job)
							}

							st, exist := cache.Cache.GetCache(job.SubTask[i].Src)
							if exist {
								status = st.(int)
							}
						} else {
							GetSlaveTaskTime(job)
							status, _ = GetResultOftheTask(job.SubTask[i])
							job.SubTask[i].Status = status
						}

						if status == TaskFinished {
							finishedMap[job.SubTask[i].Src] = TaskFinished
							cache.Cache.SetCache(job.SubTask[i].Src, TaskFinished)
							writeFlag = true
							job.SubTask[i].Status = TaskFinished
							//job.SubTask[i].Finish = time.Now().Format("2006-01-02 15:04:05")
							cache.Cache.SetCache(taskChange, job)
							seelog.Infof("Begin to finish partof %v", job)
							TaskSave(UnFinishedPrefix, job)
							seelog.Infof("job.Name%s, job.Node %s finished", job.SubTask[i].Src, job.SubTask[i].Node)
							if checkTaskFinish(finishedMap, job) {
								finished <- struct{}{}
							}
						} else if status == TaskFailed {
							job.SubTask[i].Status = TaskFailed
							cache.Cache.SetCache(job.Name, TaskFailed)
							TaskSave(UnFinishedPrefix, job)
							finished <- struct{}{}
						}

						Mutex.RLock()
						status, ok := node.NodesMap[job.SubTask[i].Node]
						if !ok || status != node.NodeAlive {
							writeFlag = true
							dest := ChooseNode(job.SubTask[i].Node, node.NodesMap)
							seelog.Infof("begin to %s change node from %s to %s", job.SubTask[i].Src, job.SubTask[i].Node, dest)
							job.SubTask[i].Node = dest
							go ExecuteTask(job.SubTask[i])
						}
						Mutex.RUnlock()
					}

					if writeFlag {
						TaskSave(UnFinishedPrefix, job)
					}
				}
			}
		}()

		<-finished
		close(done)
		ticker.Stop()

		seelog.Infof("all task finished, begin to save db...")
		status, ok := cache.Cache.GetCache(job.Name)
		if ok && status == TaskFinished {
			TaskDelete(UnFinishedPrefix, job)
		}
		seelog.Infof("begin to execute merge %d", time.Now().Unix())
		ExecuteMerge(job)
	}
}

func checkTaskFinish(finishedMap map[string]int, job *AITask) bool {
	finish := 0
	for _, value := range finishedMap {
		finish += value
	}

	if finish == len(job.SubTask)*TaskFinished {
		GetSlaveTaskTime(job)
		job.Status = TaskFinished
		TaskSave(TaskPrefix, job)

		seelog.Infof("finished all task, now: %s", time.Now().Format("2006-01-02 15:04:05"))
		return true
	}

	return false
}

func GetSlaveTaskTime(job *AITask) {
	_, finame := filepath.Split(job.Name)
	taskChange := fmt.Sprintf("%s-status", finame)
	for i := range job.SubTask {
		if job.SubTask[i].Node != job.SubTask[i].Master {
			taskTime, err := GetTimeOftheTask(job.SubTask[i])
			if err == nil {
				job.SubTask[i].PreUpBegin = taskTime.PreUpBegin
				job.SubTask[i].PreUpEnd = taskTime.PreUpEnd
				job.SubTask[i].PreUpCost = taskTime.PreUpCost
				job.SubTask[i].PreDownBegin = taskTime.PreDownBegin
				job.SubTask[i].PreDownEnd = taskTime.PreDownEnd
				job.SubTask[i].PreDownCost = taskTime.PreDownCost

				job.SubTask[i].Begin = taskTime.Begin
				job.SubTask[i].Finish = taskTime.Finish
				job.SubTask[i].Cost = taskTime.Cost

				job.SubTask[i].AfterUpBegin = taskTime.AfterUpBegin
				job.SubTask[i].AfterUpEnd = taskTime.AfterUpEnd
				job.SubTask[i].AfterUpCost = taskTime.AfterUpCost
				job.SubTask[i].AfterDownBegin = taskTime.AfterDownBegin
				job.SubTask[i].AfterDownEnd = taskTime.AfterDownEnd
				job.SubTask[i].AfterDownCost = taskTime.AfterDownCost
				cache.Cache.SetCache(taskChange, job)
			}
		}
	}
}
